package com.lzwk.app.dws.ocpx;

import com.lzwk.common.ocpx.OCPXConstant;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @title: DwsOcpxWxappMpdailyReportgroupdata
 * @Author Yeaser
 * @Date: 2022/2/14 10:15
 * @Version 1.0
 * Description: TODO
 */
public class DwsLookUpJoinTablesD {

    public static void DwsOcpxWxappMpdailyReportgroupdatas(StreamTableEnvironment tenv){
        tenv.executeSql("create table mysql_wxapp_mpdailyreportgroupdata(" +
                " `id` INT, " +
                " `account_info_id` INT, " +
                " `date` TIMESTAMP(0), " +
                " `campaign_id` STRING, " +
                " `ad_group_id` STRING, " +
                " `cost` INT, " +
                " `view_count` INT, " +
                " `valid_click_count` INT, " +
                " `ctr` DOUBLE, " +
                " `conversions_count` INT, " +
                " `conversions_cost` INT, " +
                " `conversions_rate` DOUBLE, " +
                " `official_account_follow_count` INT, " +
                " `official_account_follow_cost` INT, " +
                " `official_account_follow_rate` DOUBLE, " +
                " `order_count` INT, " +
                " `order_rate` DOUBLE, " +
                " `order_roi` DOUBLE, " +
                " `proc_time` as PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='"+ OCPXConstant.OCPXSLAVEHOST +"', " +
                "   'port'='"+ OCPXConstant.OCPXSLAVEPORT +"', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME +"', " +
                "   'password'='"+ OCPXConstant.OCPXSLAVEPASSWORD +"', " +
                "   'database-name'='"+ OCPXConstant.OCPXSLAVEDBNAME +"', " +
                "   'table-name'='"+ OCPXConstant.OCPXCDCWXAPPMPDAILYGROUPDATA +"', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '"+ OCPXConstant.SCANINCREMENTALSNAPSHOTCHUNKSIZE +"' " +
                ")");


        tenv.executeSql("create table dws_ocpx_wxapp_mpdailyreportgroupdata(" +
                "   `dt` TIMESTAMP(0), " +
                "   `account_info_id` INT, " +
                "   `appid` STRING, " +
                "   `ad_group_id` STRING, " +
                "   `ad_group_name` STRING, " +
                "   `campaign_id` STRING, " +
                "   `campaign_name` STRING, " +
                "   `account_id` STRING, " +
                "   `account_name` STRING, " +
                "   `auth_info_id` INT, " +
                "   `org_id` INT, " +
                "   `cost` INT, " +
                "   `view_count` INT, " +
                "   `valid_click_count` INT, " +
                "   `ctr` double, " +
                "   `order_count` INT, " +
                "   `order_rate` double,       " +
                "   `order_roi` double, " +
                "   `conversions_count` INT, " +
                "   `conversions_cost` INT, " +
                "   `conversions_rate` double, " +
                "   `official_account_follow_count` INT, " +
                "   `official_account_follow_cost` INT, " +
                "   `official_account_follow_rate` double, " +
                "   `id` INT,  " +
                "   PRIMARY KEY (`id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "."  + OCPXConstant.OCPXDWSWXAPPMPDAILYGROUPDATA + "', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")");


        tenv.executeSql(" insert into dws_ocpx_wxapp_mpdailyreportgroupdata " +
                " select " +
                " `t1`.`date` as dt, " +
                " `t1`.`account_info_id`, " +
                " `t6`.`appid`, " +
                " `t1`.`ad_group_id`, " +
                " regexp_replace(`t2`.`ad_group_name`, '\\n|\\t' , '') as ad_group_name, " +
                " `t1`.`campaign_id`, " +
                " regexp_replace(`t3`.`campaign_name`, '\\n|\\t' , '') as campaign_name, " +
                " `t4`.`account_id`," +
                " regexp_replace(`t4`.`account_name`, '\\n|\\t' , '') as account_name, " +
                " `t4`.`auth_info_id`, " +
                " ifnull(`t7`.`org_id`,`t8`.`org_id`) as `org_id`, " +
                " `t1`.`cost`, " +
                " `t1`.`view_count`, " +
                " `t1`.`valid_click_count`, " +
                " `t1`.`ctr`, " +
                " `t1`.`order_count`, " +
                " `t1`.`order_rate`, " +
                " `t1`.`order_roi`, " +
                " `t1`.`conversions_count`, " +
                " `t1`.`conversions_cost`, " +
                " `t1`.`conversions_rate`, " +
                " `t1`.`official_account_follow_count`, " +
                " `t1`.`official_account_follow_cost`, " +
                " `t1`.`official_account_follow_rate`, " +
                " `t1`.`id` " +
                " from mysql_wxapp_mpdailyreportgroupdata as t1 " +
                " left join ocpx_wxapp_mpadgroupdata FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                "   on t1.ad_group_id = t2.ad_group_id " +
                "   and t1.account_info_id = t2.account_info_id " +
                " left join ocpx_wxapp_mpcampaigndata FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                "   on t1.campaign_id = t3.campaign_id " +
                "   and t2.account_info_id = t3.account_info_id " +
                " left join ocpx_wxapp_mpaccountinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4 " +
                "   on t1.account_info_id = t4.id " +
                " left join ocpx_wxapp_qqauthorizationinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5" +
                "   on t4.auth_info_id = t5.id" +
                " left join ocpx_wxapp_wxapp FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t6 " +
                "   on t5.state = t6.appid " +
                " left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t7" +
                "   on t6.id = t7.object_id " +
                "   and t7.object_type = 'wxapp' " +
                " left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t8 " +
                "   on t5.id = t8.object_id " +
                "   and t8.object_type='qq_oauth'" +
                "");

/*        tenv
                .toRetractStream(table, DwsWxappMpdailyReportgroupdata.class)
                .filter(data -> data.f0)
                .map(data -> data.f1)
                .addSink(MyKafkaUtil.getKafkaProducer(OCPXConstant.DWSOCPXWXAPPMPDAILYREPORTFROUPDATA,""));*/




    }


}
